package com.gugee.ins.data.mark.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.mark.InsMarkBloggerDayDao;
import com.gugee.ins.common.dal.service.mark.InsMarkBloggerLogDao;
import com.gugee.ins.common.model.mysql.mark.InsMarkBloggerLog;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.common.msg.InsMarkMessage;
import com.gugee.ins.data.mark.config.InsMarkConfig;
import com.gugee.ins.data.mark.service.InsMarkDayService;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.ForkJoinPool;

/**
 * InsMarkDay 服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsMarkDayServiceImpl implements InsMarkDayService {

    /**
     * 配置
     */
    @Autowired
    InsMarkConfig insMarkConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * InsBloggerMainDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;

    /**
     * insMarkBloggerDayDao
     */
    @Autowired
    InsMarkBloggerDayDao insMarkBloggerDayDao;

    /**
     * insMarkBloggerLogDao
     */
    @Autowired
    InsMarkBloggerLogDao insMarkBloggerLogDao;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsUpdateEsFg;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    public static final String WORK_NAME = "insMarkDay-worker";

    /**
     *
     */
    public static final String DATE_FORMAT = "yyyyMMdd";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insMarkConfig.getInsMarkDayCorePollSize(), WORK_NAME);
        scheduleTemplate.attach(WORK_NAME, forkJoinPool);
    }

    /**
     * 处理 打点 任务
     * @param insMarkMessages
     */
    @Override
    public void handleMarkDay(List<InsMarkMessage> insMarkMessages) {
        try {
            forkJoinPool.submit(() -> insMarkMessages.parallelStream().forEach(msg -> {
                try {
                    // 打点
                    markDay(msg);
                } catch (Exception ex) {
                    log.error("MarkDay error. page:{},pageSize:{}, error :{}" ,
                            msg.getPage(),msg.getPageSize(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("MarkDay thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 打点
     *
     * @param insMarkMessage
     */
    public void markDay(InsMarkMessage insMarkMessage){
        long start = System.currentTimeMillis();
        List<Long> uids = insMarkMessage.getUids();
        if(!uids.isEmpty()){
            long now = start / 1000 - 24 * 3600;

            int today = Integer.parseInt(DateTimeUtils.formatSecond(now,DATE_FORMAT,insMarkConfig.getInsMarkDayTimeZone()));
            int yesterday = Integer.parseInt(DateTimeUtils.formatSecond(now - 24 * 3600,DATE_FORMAT,insMarkConfig.getInsMarkDayTimeZone()));
            int lastMonthDay = Integer.parseInt(DateTimeUtils.formatSecond(now - 30 * 24 * 3600,DATE_FORMAT,insMarkConfig.getInsMarkDayTimeZone()));
            int lastWeekDay = Integer.parseInt(DateTimeUtils.formatSecond(now - 7 * 24 * 3600,DATE_FORMAT,insMarkConfig.getInsMarkDayTimeZone()));
            // 插入打点表
            try{
                insMarkBloggerDayDao.insertMarkBloggerDayByUid(uids,today,yesterday,lastMonthDay,lastWeekDay);
            } catch (Exception e){
                log.error("MarkDay insert into db error. page:{},pageSize:{}, error :{}" ,
                        insMarkMessage.getPage(),insMarkMessage.getPageSize(),e.getMessage());
                // 保存打点失败日志
                saveMarkDayLog(insMarkMessage.getPage(),insMarkMessage.getPageSize(),today);
                return;
            }
            try{
                // 先更新blogger main,再更新blogger 顺序不能变
                insBloggerMainDao.updateLastFollowerGrowByUids(uids,today);
                insBloggerDao.updateLastFollowerGrowByUids(uids);

                // 发送消息更新es
                sendUpdateEsMsg(uids);
            }catch (Exception e){
                log.error("MarkDay update blogger/blogger main error, msg:{}",e.getMessage());
            }
        }
        if ((System.currentTimeMillis() - start) / 1000 > insMarkConfig.getInsMarkDayDebugTimeLimit()) {
            log.info("MarkDay page:{},pageSize:{} use time:{}s", insMarkMessage.getPage(),insMarkMessage.getPageSize()
                    ,(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 失败时 保存日志
     * @param page
     * @param pageSize
     */
    public void saveMarkDayLog(int page,int pageSize, int dateCode){
        InsMarkBloggerLog insMarkBloggerLog = new InsMarkBloggerLog();
        insMarkBloggerLog.setPage(page);
        insMarkBloggerLog.setPageSize(pageSize);
        insMarkBloggerLog.setDatecode(dateCode);
        insMarkBloggerLogDao.save(insMarkBloggerLog);
    }

    /**
     * 发送更新es的消息
     * @param uids
     */
    public void sendUpdateEsMsg(List<Long> uids){
        try{
            for(Long uid : uids){
                if(uid != null){
                    InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
                    insBloggerMessage.setUid(uid);
                    insBloggerMessage.setRetryLimitTimes(insMarkConfig.getInsBloggerEsRetryTimes());
                    insBloggerMessage.setRetryTimes(insMarkConfig.getInsBloggerEsRetryTimes());
                    kafkaTemplateInsUpdateEsFg.send(insMarkConfig.getInsMBloggerUpdateEsOnlineTopic(),insBloggerMessage);
                }
            }
        }catch (Exception e){
            log.error("MarkDay send msg to update es error, msg:{}",e.getMessage());
        }
    }
}
